package src.main.scala.com.lagou.homework.part05
import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.concurrent.duration._

/**
 * @author: yehw
 * @date: 2020/10/11 14:16
 * @description: 注册服务段
 */
//Master端
class Master extends Actor{
  //构造代码先运行
  println("Master constructor invoked")
  //定义一个map集合，用于保存worker的注册信息，key: workerId     value: WorkerInfo
       private val workerInfoMap = new mutable.HashMap[String,WorkerInfo]()

  //定义一个list集合，用于保存workerinfo
       private val workerInfoList = new ListBuffer[WorkerInfo]

  //定义master定时检查时间间隔
       val checkOutTimeInterval=15000 //15秒


  //preStart方法会在构造代码执行后，被调用，并且只会执行一次
  override def preStart(): Unit = {
    println("preStart method invoked")

    //master定时检查超时的worker
    //手动导入隐式转换
    import context.dispatcher
    context.system.scheduler.schedule(0 millis,checkOutTimeInterval millis,self,CheckOutTime)
  }

  //receive方法会在preStart方法调用后被执行，意味着可以不断的接受数据
  override def receive: Receive = {
    //接受worker的注册信息
    case RegisterMessage(workerId,memory,cores) =>{
        //判断worker是否注册，master只接受没有注册过的worker信息
        if(!workerInfoMap.contains(workerId)){
            //构建WorkerInfo对象
            val workerInfo = new WorkerInfo(workerId,memory,cores)
           //把信息保存在map集合中
            workerInfoMap +=(workerId -> workerInfo)

           //把workerinfo信息保存在list集合中
            workerInfoList +=workerInfo

          //master反馈一个注册成功信息给worker
          //通过sender方法拿到worker的引用
          sender ! RegisteredMessage(s"workerId:$workerId 注册成功")

        }
    }

       //master接受worker的心跳
    case HeartBeat(workerId) =>{
        //master要判断worker是否注册，master只接受注册过的worker的心跳信息
         if(workerInfoMap.contains(workerId)){
            //获取当前系统时间
            val now: Long = System.currentTimeMillis()

            //通过workerId获取对应的WorkerInfo信息
            val workerInfo: WorkerInfo = workerInfoMap(workerId)

           //给workerInfo中上一次心跳时间赋值
             workerInfo.lastHeartBeatTime=now

         }
    }
    //master接受自己发送过来的信息
    case CheckOutTime =>{
      //需要判断哪些worker超时
         //超时的worker判断逻辑： now - worker上一次心跳时间 > master定时检查的时间间隔 15s
      val now: Long = System.currentTimeMillis()

      val outTimeWorkerInfo: ListBuffer[WorkerInfo] = workerInfoList.filter( x=> now -x.lastHeartBeatTime >checkOutTimeInterval )

       //遍历
        for(out <- outTimeWorkerInfo){
          //从map集合中移除超时的worker信息
            val workerId: String = out.workerId
            workerInfoMap -=(workerId)
          //从list集合中移除超时的worker信息
            workerInfoList -=out

          println(s"workerId:$workerId 超时")
        }

        println("活着的worker总数："+workerInfoList.size)

       //按照worker自身的内存大小降序排列
       println(workerInfoList.sortBy(x=>x.memory).reverse)


    }
  }
}

object Master{
  def main(args: Array[String]): Unit = {
    //定义master的IP地址
    val host="127.0.0.1"

    //定义master的端口
    val port="8080"

    //准备字符串配置信息
    val configStr=
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
        """.stripMargin

    //通过ConfigFactory调用parseString解析字符串获取config对象
    val config:Config=ConfigFactory.parseString(configStr)

    //1、创建ActorSystem老大，它负责创建和监督actor
    val masterActorSystem = ActorSystem("masterActorSystem",config)

    //2、通过masterActorSystem老大构建master actor
    val masterActor: ActorRef = masterActorSystem.actorOf(Props(new Master),"masterActor")

    //3、测试
    //masterActor ! "connect"
  }
}




